package com.zfzn.flink;

import com.google.gson.Gson;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.rabbitmq.RMQSource;
import org.apache.flink.streaming.connectors.rabbitmq.common.RMQConnectionConfig;

import java.util.stream.Stream;

/**
 * @Auther lolineet
 * @Email tj.zhanghongwei@163.com
 * @Date 2021/8/19 9:56
 * @Description
 */
public class FraudDetectionJob {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        RMQConnectionConfig connectionConfig = new RMQConnectionConfig.Builder()
                .setHost("127.0.0.1")
                .setPort(5672)
                .setUserName("admin")
                .setPassword("admin")
                .setVirtualHost("/")
                .build();

        DataStream<String> dataStreamSource =
                env.addSource(new RMQSource<>(connectionConfig,"zfzn",true,new SimpleStringSchema()));
        dataStreamSource.print();

        DataStream<Transaction> transactionDataStream=dataStreamSource.map(new MapFunction<String, Transaction>() {
            @Override
            public Transaction map(String s) throws Exception {
                Gson gson = new Gson();
                Transaction transaction = gson.fromJson(s,Transaction.class);
                return transaction;
            }
        });

        DataStream<Alert> alertDataStream=transactionDataStream.keyBy(Transaction::getAccountId).process(new FraudDetector()).name("fraud-detector");
        alertDataStream.addSink(new SinkMysql());
        env.execute();

    }
}
